package com.learn.kafka.config;


import com.google.common.collect.Maps;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.Map;

/**
 * 事务配置（确保多写原子性）
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/9 10:41
 * @updateUser: MI
 * @updateTime: 2025/8/9 10:41
 * @updateRemark: 修改内容
 * @version: 1.0
 */
@Configuration
public class KafkaTransactionConfig {

    // 1. 支持事务的生产者工厂
    @Bean
    public ProducerFactory<String, Object> producerFactory(KafkaProperties properties) {

        // 获取基础配置
        Map<String, Object> configProps = Maps.newHashMap();
        configProps.put("bootstrap.servers", properties.getBootstrapServers());
        // 3. 强制设置序列化器（关键：覆盖可能的 null 值）
        // 若 properties 中未配置序列化器，这里显式指定（避免 null）
        if (!configProps.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        }
        if (!configProps.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        }
        configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-" + System.currentTimeMillis());  // 事务 ID
        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(configProps);
        factory.setTransactionIdPrefix("order-tx-prefix-");  // 事务 ID 前缀
        return factory;
    }

    // 2. 支持事务的 KafkaTemplate
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    // 3. Kafka 事务管理器（配合 @Transactional 注解）
    @Bean
    public KafkaTransactionManager<String, Object> transactionManager(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }
}
